package com.fanbo;

import com.fanbo.bean.ConsumerComponent;
import java.util.Calendar;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import org.apache.dubbo.config.spring.context.annotation.EnableDubbo;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.PropertySource;

public class DubboConsumerMain {

  public static void main(String[] args) {
    AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(DubboConsumerConfiguration.class);
    context.start();

    ConsumerComponent bean = context.getBean(ConsumerComponent.class);

    // 定时上报，每隔5s打印一次最近1分钟内每个方法的TP90、TP99的耗时情况
    ScheduledExecutorService pool = Executors.newScheduledThreadPool(5);
    pool.scheduleAtFixedRate(() -> {
      Map<String, ConcurrentHashMap<Long, Integer>> methodExecutionTime = TPMonitorFilter.METHOD_EXECUTION_TIME;

      // 1分钟之前的时间
      Calendar beforeTime = Calendar.getInstance();
      beforeTime.add(Calendar.MINUTE, -1);
      long limitTime = beforeTime.getTime().getTime();

      // 计算一分钟内的TP90、TP99
      for (Map.Entry<String, ConcurrentHashMap<Long, Integer>> entry : methodExecutionTime.entrySet()) {
        String methodName = entry.getKey();
        ConcurrentHashMap<Long, Integer> executionTimeMap = entry.getValue();
        List<Entry<Long, Integer>> currentTimeEntryList = executionTimeMap.entrySet().stream()
            .filter(e -> e.getKey() > limitTime)
            .sorted(Comparator.comparing(Map.Entry::getValue))
            .collect(Collectors.toList());
        long tp90Index = Math.round(currentTimeEntryList.size() * 0.9);
        long tp99Index = Math.round(currentTimeEntryList.size() * 0.99);
        Integer tp90 = currentTimeEntryList.get(Integer.parseInt(tp90Index + "")).getValue();
        Integer tp99 = currentTimeEntryList.get(Integer.parseInt(tp99Index + "")).getValue();
        System.out.println(methodName+":TP90->"+tp90+"毫秒，TP99->"+tp99+"毫秒");
        // 清除一分钟之前的数据
        executionTimeMap.entrySet().removeIf(e -> e.getKey() < limitTime);
      }
      System.out.println("----------------------------");
    },0,5, TimeUnit.SECONDS);

    // 获取线程池
    ThreadPoolExecutor threadPoolExecutor = getThreadPoolExecutor();
    while (true){
      threadPoolExecutor.submit(() -> {
        bean.methodA();
        bean.methodB();
        bean.methodC();
      });
      // 每间隔20ms请求一次 ，一分钟请求3000次
      try {
        Thread.sleep(20);
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    }


  }


  @Configuration
  @EnableDubbo
  @PropertySource("classpath:/dubbo-consumer.properties")
  @ComponentScan(basePackages = "com.fanbo.bean")
  static class DubboConsumerConfiguration{
  }

  public static ThreadPoolExecutor getThreadPoolExecutor(){
    int cpuCount = Runtime.getRuntime().availableProcessors();
    ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(
        Math.max(2, Math.min(cpuCount - 1, 4)),
        cpuCount * 2 + 1,
        10L,
        TimeUnit.SECONDS,
        new LinkedBlockingQueue<>(2000),
        new NamedThreadFactory("dubbo-filter-test"),
        new ThreadPoolExecutor.CallerRunsPolicy());
    return poolExecutor;
  }

  /**
   * 命名线程工厂
   */
  static class NamedThreadFactory implements ThreadFactory {

    private static final AtomicInteger poolNumber = new AtomicInteger(1);
    private final ThreadGroup group;
    private final AtomicInteger threadNumber = new AtomicInteger(1);
    private final String namePrefix;

    NamedThreadFactory(String name) {
      SecurityManager s = System.getSecurityManager();
      group = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();
      if (null == name || name.isEmpty()) {
        name = "pool";
      }
      namePrefix = "【" + name + "】-线程池编号：" + poolNumber.getAndIncrement() + "-线程编号：";
    }

    @Override
    public Thread newThread(Runnable r) {
      Thread t = new Thread(group, r, namePrefix + threadNumber.getAndIncrement(), 0);
      if (t.isDaemon()) {
        t.setDaemon(false);
      }
      if (t.getPriority() != Thread.NORM_PRIORITY) {
        t.setPriority(Thread.NORM_PRIORITY);
      }
      return t;
    }
  }


}
